"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 行存压缩
Case Name   : 普通表，非压缩表改为压缩表
Create At   : 2023/8/1
Owner       : opentestcase013
Description :
    1、建普通表并插入数据
    2、查看表文件占用磁盘大小
    3、查看oid,relfilenode
    4、依据relfilenode，查找表文件
    5、修改表为压缩表，checkpoint
    6、再次查看表文件占用磁盘大小
    7、再次查看oid,relfilenode
    8、依据oid,relfilenode，查找表文件
    9、查看文件压缩页面的管理信息
    10、查看页面trunk使用的详细信息
    11、查看压缩率
    12、清理环境
Expect      :
    1、成功
    2、成功
    3、一致
    4、成功
    5、成功
    6、包含压缩参数，表文件磁盘占用小于非压缩表
    7、不一致
    8、文件删除，新文件名加后缀_compress
    9、成功
    10、成功
    11、成功
    12、成功
History     :
"""

import os
import unittest

from yat.test import Node
from yat.test import macro

from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger


class RowCompressTest(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.pri_user = Node('PrimaryDbUser')
        self.constant = Constant()
        self.tb_name = 't_alter_rowcompress_0001'
        self.func_findpath = 'f_alter_rowcompress_0001'
        self.file_path = os.path.join(macro.DB_INSTANCE_PATH, 'base')

    def test_row_compress(self):
        text = '-----step1:建普通表并插入数据;expect:成功-----'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name};
        create table {self.tb_name}(id int, c1 text);
        insert into {self.tb_name} select generate_series(1, 1000), 
        'testrowcompress';
        '''
        sql_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn('INSERT 0 1000', sql_res, '执行失败：' + text)

        text = '-----step2:查看表文件占用磁盘大小;expect:成功-----'
        self.log.info(text)
        sql = f"select pg_size_pretty(" \
              f"pg_total_relation_size('{self.tb_name}')) as table_size;"
        sql_res = self.pri_sh.execut_db_sql(sql)
        table_size_before = sql_res.splitlines()[-2].strip()
        self.log.info(sql_res)
        self.assertNotIn('ERROR', sql_res, '执行失败：' + text)

        text = '-----step3:查看oid,relfilenode;expect:一致-----'
        self.log.info(text)
        sql = f"select oid, relfilenode from pg_class " \
              f"where relname = '{self.tb_name}';"
        sql_res = self.pri_sh.execut_db_sql(sql).splitlines()[-2].split('|')
        self.log.info(sql_res)
        sql_res = [i.strip() for i in sql_res]
        oid = sql_res[0]
        relfilenode_before = sql_res[1]
        self.assertEqual(oid, relfilenode_before, '执行失败：' + text)

        text = '-----step4:依据relfilenode，查找表文件;expect:成功-----'
        self.log.info(text)
        cmd = f'cd {self.file_path} &&' \
              f'find . -name *{relfilenode_before}*'
        cmd_res = self.pri_user.sh(cmd).result()
        self.log.info(cmd_res)
        self.assertIsNotNone(cmd_res, '执行失败：' + text)

        text = '-----step5:修改表为压缩表，checkpoint;expect:成功-----'
        self.log.info(text)
        sql = f'alter table {self.tb_name} set (compresstype = 1);' \
              f'checkpoint;'
        sql_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.ALTER_TABLE_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.CHECKPOINT_SUCCESS_MSG, sql_res,
                      '执行失败：' + text)

        text = '-----step6:再次查看表文件占用磁盘大小;' \
               'expect:包含压缩参数，表文件磁盘占用小于非压缩表-----'
        self.log.info(text)
        sql = f"select pg_size_pretty(" \
              f"pg_total_relation_size('{self.tb_name}')) as table_size;"
        sql_res = self.pri_sh.execut_db_sql(sql)
        table_size_after = sql_res.splitlines()[-2].strip()
        self.log.info(sql_res)
        self.assertGreaterEqual(table_size_before, table_size_after,
                                '执行失败：' + text)

        sql_res = self.pri_sh.execut_db_sql(f'\\d+ {self.tb_name}')
        self.log.info(sql_res)
        self.assertIn('compresstype', sql_res, '执行失败：' + text)

        text = '-----step7:再次查看oid,relfilenode;expect:不一致-----'
        self.log.info(text)
        sql = f"select oid, relfilenode from pg_class " \
              f"where relname = '{self.tb_name}';"
        sql_res = self.pri_sh.execut_db_sql(sql).splitlines()[-2].split('|')
        self.log.info(sql_res)
        sql_res = [i.strip() for i in sql_res]
        oid = sql_res[0]
        relfilenode_after = sql_res[1]
        self.assertNotEqual(oid, relfilenode_after, '执行失败：' + text)

        text = '-----step8:依据relfilenode，查找表文件expect:旧文件删除，' \
               '新文件名加后缀_compress-----'
        self.log.info(text)
        cmd = f'cd {self.file_path} &&' \
              f'find . -name *{relfilenode_before}*'
        cmd_res = self.pri_user.sh(cmd).result()
        self.log.info(cmd_res)
        self.assertEqual('', cmd_res, '执行失败：' + text)

        cmd = f'cd {self.file_path} &&' \
              f'find . -name *{relfilenode_after}*'
        cmd_res = self.pri_user.sh(cmd).result()
        self.log.info(cmd_res)
        self.assertIsNotNone(cmd_res, '执行失败：' + text)
        self.assertIn('_compress', cmd_res, '执行失败：' + text)

        text = '-----step9:查看文件压缩页面的管理信息;expect:成功-----'
        self.log.info(text)
        sql = f"select chunk_size, algorithm from " \
              f"pg_catalog.compress_address_header('{self.tb_name}', 0);"
        sql_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn('4096', sql_res, '执行失败：' + text)

        text = '-----step10:查看页面trunk使用的详细信息;expect:成功-----'
        self.log.info(text)
        sql = f"select nchunks, chunknos from " \
              f"pg_catalog.compress_address_details('{self.tb_name}', 0);"
        sql_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertNotIn('ERROR', sql_res, '执行失败：' + text)

        text = '-----step11:查看压缩率;expect:成功-----'
        self.log.info(text)
        sql = f'''drop function if exists 
        {self.func_findpath}(character varying);
        create or replace function {self.func_findpath}(character varying)
        returns character varying
        language plpgsql
        as
        \$body\$
        declare
        relpath character varying;
        begin
        relpath = (select pg_relation_filepath(relname::regclass) \
        from pg_class where relname = \$1);
        return relpath;
        end;
        \$body\$;
        select is_compress, file_count, logic_size, physic_size, 
        compress_ratio from 
        pg_catalog.compress_ratio_info({self.func_findpath}('{self.tb_name}'));
        '''
        sql_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.DROP_FUNCTION_SUCCESS_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn('3.49%', sql_res, '执行失败：' + text)
        self.assertNotIn('ERROR', sql_res, '执行失败：' + text)

    def tearDown(self):
        text = '-----step12:清理环境; expect:成功-----'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name};
        drop function if exists {self.func_findpath}(character varying);
        '''
        drop_res = self.pri_sh.execut_db_sql(sql)
        self.log.info(drop_res)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, drop_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.DROP_FUNCTION_SUCCESS_MSG,
                      drop_res, '执行失败：' + text)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')